package com.lzcu.maoyan.mapreduce.mapreduce.reduce;

import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieLeadingActorWritable;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieMonthCountWritable;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieTypeWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Arrays;

/**
 * @author WangLiang
 * @date `2024/5/8`
 */
public class MovieMonthCountReduce extends Reducer<Text, Text, NullWritable, MovieMonthCountWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        MovieMonthCountWritable movieMonthCountWritable = new MovieMonthCountWritable();
        movieMonthCountWritable.setTimeData(key.toString());
        int count = 0;
        StringBuilder movieNames = new StringBuilder();
        long total = 0;
        for (Text value : values) {
            String[] valueData = value.toString().split("\001");
            if (valueData.length == 2) {
                count += 1;
                movieNames.append(valueData[0]);
                movieNames.append("\001");
                total += Long.parseLong(valueData[1]);
            }
        }
        if (count != 0 && total != 0 && !"".equals(movieNames.toString())){
            movieMonthCountWritable.setMaxTotal(total);
            movieMonthCountWritable.setMovieName(movieNames.toString());
            movieMonthCountWritable.setCount(count);
            context.write(NullWritable.get(), movieMonthCountWritable);
        }
    }
}
